package net.bw.realtime.jtp.dws.log.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

/**
 * @author liuyawei
 * @date 2025/5/21
 */
public class JtpTrafficAppStartMinuteWindowDwsJob {

    public static void main(String[] args) {
        
        // 1. 创建执行环境
        TableEnvironment tabEnv = getTableEnv();

        // 2. 创建输入表,映射到kafka消息队列
        createInputTable(tabEnv);

        // 3. 处理数据
        handle(tabEnv);

        // 4.创建输出表,映射到clickhouse
        createOutputTable(tabEnv);

        // 5.保存数据
        saveToClickHouse(tabEnv);

        // 6.关联广告维表数据
        joinDim(tabEnv);

    }

    private static void joinDim(TableEnvironment tabEnv) {

        // App启动广告统计
        Table ad_start_table = tabEnv.sqlQuery(
                "SELECT " +
                        " window_start AS window_start_time" +
                        " , window_end AS window_end_time" +
                        " , open_ad_id\n" +
                        " , count(open_ad_id) AS open_ad_count\n" +
                        " , SUM(open_ad_ms) AS open_ad_ms\n" +
                        " , SUM(open_ad_skip_ms) AS open_ad_skip_ms\n" +
                        " , UNIX_TIMESTAMP() * 1000 AS ts\n" +
                        " , PROCTIME() AS proc_time\n" +
                        " FROM TABLE(\n" +
                        "    TUMBLE(TABLE start_log_table, DESCRIPTOR(row_time), INTERVAL '1' MINUTES))\n" +
                        "  GROUP BY window_start, window_end, open_ad_id"
        );

        tabEnv.createTemporaryView("ad_start_result_table", ad_start_table);

        // 创建维表 lookup join方式
        tabEnv.executeSql(
                "CREATE TABLE `ads_info_mysql_source_dim`\n" +
                        "(\n" +
                        "   id                STRING COMMENT '主键ID',\n" +
                        "   name              STRING COMMENT '广告名称',\n" +
                        "   content           STRING COMMENT '广告内容(图片URL/视频URL/H5代码)',\n" +
                        "   type              STRING COMMENT '广告类型(IMAGE, VIDEO, HTML5)',\n" +
                        "   start_time        TIMESTAMP COMMENT '展示开始时间',\n" +
                        "   end_time          TIMESTAMP COMMENT '展示结束时间',\n" +
                        "   is_active         INT COMMENT '是否启用',\n" +
                        "   target_condition  STRING COMMENT '定向条件（JSON格式）',\n" +
                        "   created_at        TIMESTAMP COMMENT '数据插入时间',\n" +
                        "   updated_at        TIMESTAMP COMMENT '数据更新时间',\n" +
                        "   PRIMARY KEY (id) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "   'connector' = 'jdbc',\n" +
                        "   'url' = 'jdbc:mysql://node103:3306/jtp_ads',\n" +
                        "   'table-name' = 'ads_info', \n" +
                        "   'lookup.cache.max-rows' = '10', \n" +
                        "   'lookup.cache.ttl' = '1 hour', \n" +
                        "   'username' = 'root', \n" +
                        "   'password' = '123456', \n" +
                        "   'driver' = 'com.mysql.cj.jdbc.Driver' \n" +
                        ")"
        );

        // 汇总数据
        Table joinTable = tabEnv.sqlQuery(
                "SELECT\n" +
                        " t1.window_start_time\n" +
                        " , t1.window_end_time\n" +
                        " , t2.name AS open_ad_name\n" +
                        " , t1.open_ad_count\n" +
                        " , t1.open_ad_ms\n" +
                        " , t1.open_ad_skip_ms\n" +
                        " , t1.ts\n" +
                        "FROM ad_start_result_table t1\n" +
                        "LEFT JOIN ads_info_mysql_source_dim FOR SYSTEM_TIME AS OF t1.proc_time AS t2 \n" +
                        "ON CAST(t1.open_ad_id AS STRING) = t2.id"
        );

        joinTable.execute().print();

    }

    // 保存数据
    private static void saveToClickHouse(TableEnvironment tabEnv) {

        // 启动APP不同方式窗口汇总表
        tabEnv.executeSql(
                "INSERT INTO dws_traffic_app_start_window_report_clickhouse_sink\n" +
                        "SELECT\n" +
                        "    DATE_FORMAT(window_start_time, 'yyyy-MM-dd HH:mm:ss') AS window_start_time\n" +
                        "    , DATE_FORMAT(window_end_time, 'yyyy-MM-dd HH:mm:ss') AS window_end_time\n" +
                        "    , entry\n" +
                        "    , entry_count\n" +
                        "    , loading_time\n" +
                        "    , ts\n" +
                        "FROM start_report_table"
        );

        //启动APP展示广告窗口汇总表
        tabEnv.executeSql(
                "INSERT INTO dws_traffic_app_start_ad_window_report_clickhouse_sink\n" +
                        "SELECT\n" +
                        "    DATE_FORMAT(window_start_time, 'yyyy-MM-dd HH:mm:ss') AS window_start_time\n" +
                        "     , DATE_FORMAT(window_end_time, 'yyyy-MM-dd HH:mm:ss') AS window_end_time\n" +
                        "     , open_ad_id\n" +
                        "     , open_ad_count\n" +
                        "     , open_ad_ms\n" +
                        "     , open_ad_skip_ms\n" +
                        "     , ts\n" +
                        "FROM start_ad_report_table"
        );

    }

    // 创建输出表,映射到clickhouse
    private static void createOutputTable(TableEnvironment tabEnv) {

        // 创建APP启动类型统计输出表
        tabEnv.executeSql(
                "CREATE TABLE dws_traffic_app_start_window_report_clickhouse_sink (\n" +
                        "  `window_start_time` STRING COMMENT '窗口开始日期时间',\n" +
                        "  `window_end_time` STRING COMMENT '窗口结束日期时间',\n" +
                        "  `entry` STRING COMMENT '启动app方式',\n" +
                        "  `entry_count` BIGINT COMMENT '启动app方式次数',\n" +
                        "  `loading_time` BIGINT COMMENT '启动app方式次数',\n" +
                        "  `ts` BIGINT COMMENT '启动app加载时间'\n" +
                        ") WITH (\n" +
                        "  'connector' = 'clickhouse',\n" +
                        "  'url' = 'jdbc:clickhouse://node103:8123/jtp_log_report',\n" +
                        "  'table' = 'dws_traffic_app_start_window_report',\n" +
                        "  'username' = 'default',\n" +
                        "  'password' = '',\n" +
                        "  'format' = 'json'\n" +
                        ")"
        );

        // 创建APP启动广告统计输出表
        tabEnv.executeSql(
                "CREATE TABLE dws_traffic_app_start_ad_window_report_clickhouse_sink (\n" +
                        "  `window_start_time` STRING COMMENT '窗口开始日期时间',\n" +
                        "  `window_end_time` STRING COMMENT '窗口结束日期时间',\n" +
                        "  `open_ad_id` STRING COMMENT '启动app方式',\n" +
                        "  `open_ad_count` BIGINT COMMENT '广告展示次数',\n" +
                        "  `open_ad_ms` BIGINT COMMENT '广告总共播放时间',\n" +
                        "  `open_ad_skip_ms` BIGINT COMMENT '用户跳过广告时间',\n" +
                        "  `ts` BIGINT COMMENT '数据计算时间戳'\n" +
                        ") WITH (\n" +
                        "  'connector' = 'clickhouse',\n" +
                        "  'url' = 'jdbc:clickhouse://node103:8123/jtp_log_report',\n" +
                        "  'table' = 'dws_traffic_app_start_ad_window_report',\n" +
                        "  'username' = 'default',\n" +
                        "  'password' = '',\n" +
                        "  'format' = 'json'\n" +
                        ")"
        );

    }

    // 处理数据
    private static void handle(TableEnvironment tabEnv) {

        // 提取启动日志字段信息
        Table start_log_table = tabEnv.sqlQuery(
                "SELECT " +
                        "`start`['entry'] AS entry," +
                        " CAST(`start`['loading_time'] AS BIGINT) AS loading_time," +
                        " `start`['open_ad_id'] AS open_ad_id," +
                        " CAST(`start`['open_ad_ms'] AS BIGINT) AS open_ad_ms," +
                        " CAST(`start`['open_ad_skip_ms'] AS BIGINT) AS open_ad_skip_ms," +
                        " row_time" +
                        " FROM dwd_traffic_start_log_kafka_source" +
                        " WHERE `start`['entry'] IS NOT NULL"
        );

        // 创建临时视图
        tabEnv.createTemporaryView("start_log_table", start_log_table);

        // APP启动类型统计
        Table start_report_table = tabEnv.sqlQuery(
                "SELECT\n" +
                        "  TUMBLE_START(row_time, INTERVAL '1' MINUTES) AS window_start_time,\n" +
                        "  TUMBLE_END(row_time, INTERVAL '1' MINUTES) AS window_end_time,\n" +
                        "  entry,\n" +
                        "  count(entry) AS entry_count,\n" +
                        "  sum(loading_time) AS loading_time,\n" +
                        "  UNIX_TIMESTAMP() * 1000 AS ts\n" +
                        "  FROM start_log_table\n" +
                        "GROUP BY\n" +
                        "  TUMBLE(row_time, INTERVAL '1' MINUTES),\n" +
                        "  entry"
        );

        // todo 使用Windowing TVF表值函数进行计算
//        tabEnv.sqlQuery(
//                "SELECT " +
//                        " window_start AS window_start_time" +
//                        " , window_end AS window_end_time" +
//                        " , entry\n" +
//                        " , count(entry) AS entry_count\n" +
//                        " , SUM(loading_time) AS loading_time\n" +
//                        " , UNIX_TIMESTAMP() * 1000 AS ts\n" +
//                        "  FROM TABLE(\n" +
//                        "    TUMBLE(" +
//                        "       DATA => TABLE start_log_table,\n" +
//                        "       TIMECOL => DESCRIPTOR(row_time),\n" +
//                        "       SIZE => INTERVAL '1' MINUTES" +
//                        "   )" +
//                        ")\n" +
//                        "  GROUP BY window_start, window_end, entry"
//        );

        // 创建临时视图、
        tabEnv.createTemporaryView("start_report_table", start_report_table);

        // APP启动广告统计
        Table start_ad_report_table = tabEnv.sqlQuery(
                "SELECT\n" +
                        "  TUMBLE_START(row_time, INTERVAL '1' MINUTES) AS window_start_time,\n" +
                        "  TUMBLE_END(row_time, INTERVAL '1' MINUTES) AS window_end_time,\n" +
                        "  open_ad_id,\n" +
                        "  count(open_ad_id) AS open_ad_count,\n" +
                        "  sum(open_ad_ms) AS open_ad_ms,\n" +
                        "  sum(open_ad_skip_ms) AS open_ad_skip_ms,\n" +
                        "  UNIX_TIMESTAMP() * 1000 AS ts\n" +
                        "  FROM start_log_table\n" +
                        "GROUP BY\n" +
                        "  TUMBLE(row_time, INTERVAL '1' MINUTES),\n" +
                        "  open_ad_id"
        );

        // 创建临时视图
        tabEnv.createTemporaryView("start_ad_report_table", start_ad_report_table);

    }

    // 创建输入表
    private static void createInputTable(TableEnvironment tabEnv) {

        // 映射到kafka消息队列
        tabEnv.executeSql("CREATE TABLE dwd_traffic_start_log_kafka_source\n" +
                "(\n" +
                "    common     MAP<STRING, STRING> COMMENT '公共环境信息',\n" +
                "    `start`    MAP<STRING, STRING> COMMENT '启动信息',\n" +
                "    ts         BIGINT,\n" +
                "    row_time   AS TO_TIMESTAMP_LTZ(ts, 3),\n" +
                "    WATERMARK FOR row_time AS row_time - INTERVAL '0' MINUTE\n" +
                ") WITH (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'dwd-traffic-start-log',\n" +
                "    'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                "    'properties.group.id' = 'gid_dws_log_app_start',\n" +
                "    'scan.startup.mode' = 'earliest-offset',\n" +
                "    'format' = 'json',\n" +
                "    'json.fail-on-missing-field' = 'false',\n" +
                "    'json.ignore-parse-errors' = 'true'\n" +
                ")"
        );

    }


    // 创建执行环境
    private static TableEnvironment getTableEnv() {

        //  1. 创建表执行环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();

        TableEnvironment tabEnv = TableEnvironment.create(settings);

        // 2. 配置属性
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");

        return tabEnv;
    }

}
